package zhanghao;



import java.io.IOException;



import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;






import java.util.regex.Matcher;
import java.util.regex.Pattern;

import net.sf.json.JSONObject;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.bj58.analysis.tools.CommonAnalysisToolsDriver;
import com.bj58.analysis.tools.LogToolsTag;
import com.bj58.analysis.tools.etl.bean.ETLTrackBean;
import com.bj58.analysis.track.format.TrackSplitUtils;


public class SourceAllFlowlMap  extends Mapper<LongWritable,Text, Text, Text>{

	private boolean filterSource(String source){
		
		if(null == source || StringUtils.isBlank(source)){
			return false;
		}
		
		if(source.length() > 80 || source.startsWith("http://")){
			return false;
		}
		
		if(source.contains("\\\\x") || source.contains("\\x") || source.contains("*") || source.contains("alert") || source.contains("<script>") ){
			return false;
		}
		
		String reg = "[\u4E00-\u9FA5${}\t\r\n\\s]+";
		Pattern p = Pattern.compile(reg);
        Matcher m = p.matcher(source);
        if(m.find()){
			return false;
		}
        
        String regd = "[^.]";
		Pattern pd = Pattern.compile(regd);
        Matcher md = pd.matcher(source);
        if(!md.find()){
			return false;
		}
		
		
		return true;
	}
	
	
	Text outkey = new Text();
	Text outvalue = new Text();
	TrackSplitUtils trackSplit = TrackSplitUtils.getInstance();
	private String fenge = "::";
	private int regionsnum = 10;
	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		
		String date = context.getConfiguration().get("DATE");
		String prefix = "0" + (Integer.parseInt(date.substring(5))%regionsnum);
//		regionsnum
		
		String platform = context.getConfiguration().get("PLATFORM");
		
		
		
		// 用于存储汇总后rawkey 和 小范围pv
		Map<String,Long> rowKey_Entry = new HashMap<String,Long>();
		// 开始处理业务
		List<List<String>> listLines = trackSplit.getSplitVisit(value);
		for(List<String> listline : listLines){
			// session为单位   存储所有遇见过的来源（站内站外）
			Set<String> srcSet = new HashSet<String>();
			srcSet.clear();
			
			for (String line : listline) {
				try {
					ETLTrackBean trackinfo = (ETLTrackBean) CommonAnalysisToolsDriver
							.parserToInstance( line ,
									LogToolsTag.TRACK_ETL_LOG);
					
					if(null == trackinfo){
						continue;
					}
					
					String url = trackinfo.getUrl();
					
					//提取站内来源
					if (url.contains("?from=") || url.contains("&from=")) {
						String[] tmp = url.split("[?&]from=", -1);
						if (StringUtils.isNotBlank(tmp[1])) {
							try{
								String source = tmp[1].split("&")[0];
								
								if(StringUtils.isNotBlank(source)){
									source = source.split("\\?")[0];
								}
								
								if(filterSource(source)){
									srcSet.add(source);
								}
							}catch(Exception e){
								
							}
							
						}
					}
					
					
					// 提取站外来源
					if (url.contains("?utm_source=") || url.contains("&utm_source=")) {
						String[] tmp = url.split("[?&]utm_source=", -1);
						if (StringUtils.isNotBlank(tmp[1])) {
							try{
								String source = tmp[1].split("&")[0];
								
								if(StringUtils.isNotBlank(source)){
									source = source.split("\\?")[0];
								}							
								if(filterSource(source)){
									srcSet.add(source);
								}
							}catch(Exception e){
								
							}
						}
					}
					
					
					String pagetype = trackinfo.getPageType();
					if(StringUtils.isBlank(pagetype) || null == pagetype){
						pagetype = "-";
					}
					JSONObject trackUrl = JSONObject.fromObject(trackinfo.getTrackUrl());
					String page = "-";
					if (trackUrl.containsKey("page") && StringUtils.isNotBlank(trackUrl.getString("page"))) {
						page = trackUrl.getString("page");
					}
					
					String cate1 = (null == trackinfo.getCatePathOfLevel(1))?"-":trackinfo.getCatePathOfLevel(1);
					String cate2 = (null == trackinfo.getCatePathOfLevel(2))?"-":trackinfo.getCatePathOfLevel(2);

					
					String[] pagetypeArr = {pagetype + fenge + page, pagetype + fenge + "ALL", "ALL" + fenge + "ALL"};
					String[] cateArr = {cate1 + fenge + cate2, cate1 + fenge + "ALL", "ALL" + fenge + "ALL"};
					
					long pv = 0;
					
					
					// 各级汇总
					for (String source : srcSet) {
						for(String pagetypeStr : pagetypeArr){
							for(String cateStr : cateArr){
								String keyStr = prefix + fenge + date + fenge + platform + fenge + source + fenge + pagetypeStr + fenge + cateStr;
								if(rowKey_Entry.containsKey(keyStr)){
									pv = rowKey_Entry.get(keyStr)+1;
									rowKey_Entry.put(keyStr, pv);
								}else{
									rowKey_Entry.put(keyStr, 1l);
								}
							}
						}
					}
					
				}catch(Exception e){
					
				}

			}
		}
		
		// add
		for (Map.Entry<String, Long> entry : rowKey_Entry.entrySet()){
			outkey.set(entry.getKey());
			outvalue.set("1" + "\001" + entry.getValue());
			context.write(outkey, outvalue);
		}
		//end
			
	}
}
